package cn.zwx.business.order.mq.producer;

import cn.zwx.base.enums.MqTopicEnum;
import cn.zwx.business.order.constants.CommonConstant;
import cn.zwx.business.order.service.OrderService;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

/**
 * @description: 分布式消息生产者
 * @projectName:rocket-mq-example
 * @see:cn.zwx.business.order.mq
 * @author:zhangwenxue
 * @createTime:2020/10/15 12:53
 * @version:1.0
 */
@Component
public class TransactionProducer {

    private static final Logger logger =  LoggerFactory.getLogger(TransactionProducer.class);

//
//    /**
//     * 需要自定义事务监听器 用于 事务的二次确认 和 事务回查
//     */
//    private TransactionListener transactionListener ;

    /**
     * 这里的生产者和之前的不一样
     */
    private TransactionMQProducer producer = null;

    /**
     * 官方建议自定义线程 给线程取自定义名称 发现问题更好排查
     */
    private ExecutorService executorService = new ThreadPoolExecutor(16, 20, 100, TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        }

    });

    /**
     * @description 构建producer
     **/
    public TransactionProducer() {
//        transactionListener = new TransactionListenerImplForOrder(orderService);
        // 初始化 事务生产者
        producer = new TransactionMQProducer(MqTopicEnum.ORDER_TOPIC.getTopic());
        // 添加服务器地址
        producer.setNamesrvAddr(CommonConstant.NAME_SERVER);
        //生产组设置
        producer.setProducerGroup(CommonConstant.APPLICATION_NAME);
        // 添加事务监听器
//        producer.setTransactionListener(transactionListener);
        // 添加自定义线程池
        producer.setExecutorService(executorService);
        //一个实例只能启动一次
        logger.info("构建producer成功!!!");
        start();
    }


    public TransactionMQProducer getProducer() {
        return this.producer;
    }

    /**
     * 对象在使用之前必须要调用一次，只能初始化一次
     */
    public void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutdown() {
        this.producer.shutdown();
    }

}
